package com.gugee.ins.data.mark.listener;

import com.gugee.ins.data.common.consumer.AbstractGugeeConsumerHandler;
import com.gugee.ins.data.common.msg.InsMarkMessage;
import com.gugee.ins.data.mark.config.InsMarkConfig;
import com.gugee.ins.data.mark.service.InsMarkDayService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.List;


/**
 * InsAnalysisAvatar Listener
 *
 * @author Huangxin
 */
@Component
@Slf4j
public class InsMarkDayListener {

    /**
     * InsMarkConfig
     */
    @Autowired
    private InsMarkConfig insMarkConfig;

    /**
     * insBloggerFollowerService
     */
    @Autowired
    private InsMarkDayService insMarkDayService;


    private InsMarkDayHandler markHandler = new InsMarkDayHandler();

    /**
     * 监听任务
     * @param messages
     * @param ack
     */
    @KafkaListener(id="${kafka.consumer.group.insMarkDay.id}",topics = {"${kafka.topics.insMarkDay.online}"},containerFactory = "factoryInsMarkDayAck")
    public void markBloggerDay(List<InsMarkMessage> messages, Acknowledgment ack) {
        markHandler.handle("InsMarkDay",messages, insMarkConfig.getAcceptInsMarkDayMsgTimeMinRange(),ack);
    }

    class InsMarkDayHandler extends AbstractGugeeConsumerHandler<InsMarkMessage>{
        @Override
        public void dealMsg(List<InsMarkMessage> messages) {
            insMarkDayService.handleMarkDay(messages);
        }
    }
}
